/**
 * File: BucketLoadProcessor.java
 * Created by: mhaimel
 * Created on: 1 Sep 2009
 * CVS:  $Id: BucketLoadProcessor.java,v 1.1 2009/10/09 15:14:58 mhaimel Exp $
 */
package uk.ac.ebi.curtain.processor.writer.impl;

import java.io.BufferedReader;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;

import uk.ac.ebi.curtain.processor.AbstractProcessor;
import uk.ac.ebi.curtain.processor.writer.BucketCollectionBean;
import uk.ac.ebi.curtain.storage.sequence.SequenceBox;
import uk.ac.ebi.curtain.storage.sequence.fast.FastSequenceBoxReader;
import uk.ac.ebi.curtain.utils.CurtainUncheckedException;
import uk.ac.ebi.curtain.utils.concurrent.UtilsCollect;
import uk.ac.ebi.curtain.utils.data.ReadType;
import uk.ac.ebi.curtain.utils.file.EntryMapperBean;
import uk.ac.ebi.curtain.utils.file.FileInfo;
import uk.ac.ebi.curtain.utils.file.WriteUtil;
import uk.ac.ebi.curtain.utils.io.impl.FileIO;
import uk.ac.ebi.velvet.model.Supercontig;

/**
 * @author mhaimel
 *
 */
public class BucketLoadProcessor extends AbstractProcessor<File, BucketCollectionBean> implements Closeable {
	
	private final ConcurrentMap<File, FastSequenceBoxReader> reader = UtilsCollect.newConcurrentMap();
	
	@Override
	protected BucketCollectionBean buildReturnObject() {
		BucketCollectionBean bean = new BucketCollectionBean(getInput());
		return bean;
	}
	
	@Override
	public void close() {
		while(!reader.isEmpty()){
			File file = reader.keySet().toArray(new File[0])[0];
			FastSequenceBoxReader in = reader.remove(file);
			FileIO.closeQuietly(in);
		}
	}

	@Override
	public void process() {
		File baseDir = getInput();
		getLog().debug("Fill reads for bucket " + baseDir + " ...");
		List<Long> ctgIDs = loadContigs(baseDir);
		List<Long> arcIDs = loadArcIds(baseDir);
		Set<Integer> categories = getContext().getArguments().getCategories();
		for(Integer cat : categories){
			FileInfo single = Supercontig.getPairedReadOutputFile(baseDir, cat);
			if(single.getFile().exists() && !single.getFile().isDirectory()){
				FileUtils.deleteQuietly(single.getFile());
			}
			for(ReadType rt : ReadType.getAllRead()){
				if(ReadType.Contig.equals(rt)){
					continue; // Don't use Contig reads at this point
				}
				File full = getContext().currentFiles().getSequenceBoxMapping(rt, cat, true);
				File linked = getContext().currentFiles().getSequenceBoxMapping(rt, cat, false);
				FileInfo fi = Supercontig.getPairedReadOutputFile(baseDir, cat);
				boolean append = false;
				if(rt.isIn(ReadType.getUnpaired())){
					append = true;
					fi = Supercontig.getReadOutputFile(baseDir);
				}
				if(full.exists() || linked.exists()){
					PrintWriter pw = null;
					WriteUtil writer = fi.getFileType().getWriter();
					try {
						pw = new FileIO(fi.getFile(),append).getPrintWriter();
						collect(full,ctgIDs,true, writer,pw);
						collect(linked,arcIDs,false, writer,pw);
					} catch (IOException e) {
						throw new CurtainUncheckedException(e);
					}finally{
						FileIO.closeQuietly(pw);
					}
				}
			}
		}
		getLog().debug("Filling reads for bucket " + baseDir + " done!");
	}

	private void collect(File input,List<Long> ids, boolean isfull, WriteUtil writer, PrintWriter pw) throws IOException {
		if(!input.exists()){
			return;
		}
		FastSequenceBoxReader in = getOpenReader(input);
		for(Long id : ids){
			for(SequenceBox sb : in.getEntries(id)){
				for(String seq : sb.getSequences()){
					EntryMapperBean bean = new EntryMapperBean();
					bean.setId(sb.getId());
					bean.setSequence(seq);
					writer.write(pw, bean);
				}
			}
		}
	}
	private FastSequenceBoxReader getOpenReader(File input) throws IOException {
		FastSequenceBoxReader in = reader.get(input);
		if(null == in){
			in = new FastSequenceBoxReader(input);
			reader.put(input, in);
			in.open();
		}
		return in;
	}
	private List<Long> loadArcIds(File input) {
		return loadIds(Supercontig.getArcIdOutputFile(input));
	}
	private List<Long> loadContigs(File input) {
		return loadIds(Supercontig.getContigIdOutputFile(input));
	}
	
	private List<Long> loadIds(File contigIdFile) {
		if(!contigIdFile.exists()){
			String msg = "Expected file does not exist: ";
			getLog().error(msg + contigIdFile);
			throw new CurtainUncheckedException(msg);
		}
		List<Long> contigIds = new LinkedList<Long>();
		BufferedReader in = null;
		try {
			in = new FileIO(contigIdFile).getBufferedReader();
			String line = null;
			while((line = in.readLine()) != null){
				line = line.trim();
				if(StringUtils.isNotBlank(line)){
					contigIds.add(Long.valueOf(line));
				}
			}
		} catch (IOException e) {
			throw new CurtainUncheckedException("Problems reading file " + contigIdFile,e);
		}finally{
			FileIO.closeQuietly(in);
		}
		Collections.sort(contigIds);
		return contigIds;
	}


}
